set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode='nonstrict';
set hive.exec.max.dynamic.partitions=400;
set hive.exec.max.dynamic.partitions.pernode=400;

insert overwrite table jms_dm.dm_bagging_overdue_detail_dt partition(dt='{{ execution_date | cst_ds }}')
select
      send.waybill_no         as waybill_no              --运单号
     ,send.package_code       as package_code            --包号
     ,case when der.order_source_code in ('D573') then order_source_code
           when der.order_source_code in ('D67','1085') then order_source_code
           when der.order_source_code in ('D69','D60','D09') then order_source_code
           else '其他' end as order_source_code
     ,case when der.order_source_code in ('D573') then '七星潭'
           when der.order_source_code in ('D67','1085') then '紫金山'
           when der.order_source_code in ('D69','D60','D09') then '桃花岛'
           else '其他' end as order_source_name
     ,dim.manage_code         as manage_code             --管理大区编码
     ,dim.manage_name         as manage_name             --管理大区名称
     ,dim.agent_code          as bagg_agent_code         --建包网点所属代理区编码
     ,dim.agent_name          as bagg_agent_name         --建包网点所属代理区名称
     ,send.scan_site_code     as send_scan_site_code     --装车发件网点编码
     ,send.scan_site_name     as send_scan_site_name     --装车发件网点名称
     ,send.scan_time          as send_scan_time          --装车发件扫描时间
     ,send.input_time         as send_input_time         --运单号上传时间
     ,bagg.input_time         as bagg_input_time         --包牌号上传时间
     ,send.next_station_code  as send_next_station_code  --装车发件下一站编码
     ,send.next_station_name  as send_next_station_name  --装车发件下一班名称
     ,bagg.scan_site_code     as bagg_scan_site_code     --建包组织名称
     ,bagg.scan_site          as bagg_scan_site_name     --建包组织名称
from (
     select *
     from(
         select
               waybill_no         as waybill_no
              ,package_code       as package_code
              ,input_time         as input_time
              ,scan_time          as scan_time
              ,scan_site_code     as scan_site_code
              ,input_site         as scan_site_name
              ,next_station_code  as next_station_code
              ,next_station       as next_station_name
              ,auto_produce_type  as lockflag
              ,row_number() over(partition by waybill_no,scan_site_code order by scan_time desc) as rn  --去重
         from jms_dwd.dwd_tab_barscan_send_base_hi
         where substr(dt,1,10) = date_add('{{ execution_date | cst_ds }}',1)
           and auto_produce_type = 2
         ) a where a.rn = 1
    ) send
left join(
    select *
    from(
        select
               waybill_no       as waybill_no
              ,input_time       as input_time
              ,scan_site        as scan_site
              ,scan_site_code   as scan_site_code
              ,scan_time        as scan_time
              ,package_code     as package_code
              ,row_number() over(partition by waybill_no,scan_site_code order by scan_time desc) as rn  --去重
        from jms_dwd.dwd_tab_barscan_bagging_base_hi
        where substr(dt,1,10) between date_sub('{{ execution_date | cst_ds }}',1) and date_add('{{ execution_date | cst_ds }}',1)
    ) a where a.rn = 1
) bagg on send.package_code = bagg.waybill_no
left join (
    select
          waybill_id        as waybill_no
         ,order_source_code as order_source_code
         ,order_source_name as order_source_name
    from jms_dwd.dwd_yl_oms_oms_order_dt
    where dt between date_sub('{{ execution_date | cst_ds }}',21) and '{{ execution_date | cst_ds }}'
) der on send.waybill_no = der.waybill_no
left join jms_dim.dim_network_whole_massage dim
   on bagg.scan_site_code = dim.code
distribute by 1;
